package com.wy.scs.rabbitmq.consumer.listener;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.wy.scs.rabbitmq.consumer.message.MessageData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author wangyong
 */
@Slf4j
@Component
public class DemoConsumer {

    @StreamListener(MySink.DEMO_INPUT)
    public void onMessage(@Payload MessageData message,
                          @Header(AmqpHeaders.CHANNEL) Channel channel,
                          @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        log.info("收到消息：线程编号:{} deliveryTag：{} 消息内容：{}", Thread.currentThread().getId(), deliveryTag, JSON.toJSONString(message));

        // 提交消费进度
        // ack 确认消息
        // 第二个参数 multiple ，用于批量确认消息，为了减少网络流量，手动确认可以被批处。
        // 1. 当 multiple 为 true 时，则可以一次性确认 deliveryTag 小于等于传入值的所有消息
        // 2. 当 multiple 为 false 时，则只确认当前 deliveryTag 对应的消息
//        channel.basicAck(deliveryTag, true);

        // nack
        // 第一个参数 是deliveryTag
        // 第二个参数：批量参数 同上
        // 第三个参数： 是否重新放到队列中  默认会重复放回队列5次
        channel.basicNack(deliveryTag, false, true);
    }

}
